package test.redis.tool;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisSentinelPool;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;
import redis.clients.jedis.Tuple;

public class RedisConnect {
    public static RedisConnect redisInstance = new RedisConnect();

    //private static final String CONF = "napm-common.properties";
    private static Logger logger = Logger.getLogger(RedisConnect.class);
    private static JedisSentinelPool jedisPool = null;
    private static final int DEFAULT_COUNT = 10;

    static {
        logger.info("JedisPool is null and build it now");
        try {
            //jedisPool = RedisConfig.build(CONF);
            jedisPool = RedisConfig.build();
        } catch (IOException e) {
            logger.error("NCR connect error cause:", e);
        } catch (Exception e) {
            logger.error("NCR connect error cause:", e);
        }
    }

    private RedisConnect() {
        // nothing
    }

    public static RedisConnect getRedisInstance() {
        return redisInstance;
    }

    public Jedis getResource() {
        return jedisPool.getResource();
    }

    /**
     * @Description release resource
     */
    private void release(Jedis jedis) {
        if (null != jedis) {
            jedis.close();
        }
    }

    /**
     * @Description 模板方法，获取连接，执行相关操作，关闭连接
     */
    public <T> T execute(JedisCallback<T> action) {
        Jedis jedis = null;
        try {
            jedis = getResource();
            T result = action.doInRedis(jedis);
            return result;
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            return null;
        } finally {
            release(jedis);
        }
    }

    /**
     * @Description delete key
     */
    public Long delete(final String key) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.del(key);
            }
        });
    }

    /**
     * @Description batch delete keys
     */
    public void delete(final Iterable<String> keys) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                for (String key : keys) {
                    pipeline.del(key);
                }
                pipeline.sync();
                return null;
            }
        });
    }

    /**
     * @Description 判断key是否存在
     */
    public boolean exists(final String key) {
        return execute(new JedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(Jedis jedis) {
                return jedis.exists(key);
            }
        });
    }

    /**
     * @Description 判断key是否存在
     */
    public boolean exists(final byte[] key) {
        return execute(new JedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(Jedis jedis) {
                return jedis.exists(key);
            }
        });
    }

    /**
     * @Description 存储对象
     */
    public void set(final String key, final Object value) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                jedis.set(getKeyBytes(key), SerializeUtil.serialize(value));
                return null;
            }
        });
    }

    public void set(final String key, final String value) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                jedis.set(key, value);
                return null;
            }
        });
    }

    public String set(final String key, final String value, final String nxxx, final String expx, final long time) {
        return execute(new JedisCallback<String>() {
            @Override
            public String doInRedis(Jedis jedis) {
                return jedis.set(getKeyBytes(key), getKeyBytes(value), getKeyBytes(nxxx), getKeyBytes(expx), time);
            }
        });
    }

    public void setex(final String key, final int exp, final String value) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                jedis.setex(key, exp, value);
                return null;
            }
        });
    }

    /**
     * @Description 存储对象（包括有效期）
     */
    public void setex(final String key, final int exp, final Object value) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                jedis.setex(getKeyBytes(key), exp, SerializeUtil.serialize(value));
                return null;
            }
        });
    }

    public void set(final String[] keys, final String[] values) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                if (keys.length != values.length) {
                    return null;
                }
                Pipeline pipeline = jedis.pipelined();
                for (int i = 0; i < keys.length; i++) {
                    pipeline.set(keys[i], values[i]);
                }
                pipeline.sync();
                return null;
            }
        });
    }

    /**
     * @Description Batch set keys, values with expiry time
     * @param keys
     * @param exp
     * @param values
     */
    public void setex(final String[] keys, final int exp, final String[] values) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                if (keys.length != values.length) {
                    return null;
                }
                Pipeline pipeline = jedis.pipelined();
                for (int i = 0; i < keys.length; i++) {
                    pipeline.setex(keys[i], exp, values[i]);
                }
                pipeline.sync();
                return null;
            }
        });
    }

    /**
     * @Description 获取对象
     */
    public Object get(final String key) {
        return execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                return SerializeUtil.unserialize(jedis.get(getKeyBytes(key)));
            }
        });
    }

    public String getString(final String key) {
        return execute(new JedisCallback<String>() {
            @Override
            public String doInRedis(Jedis jedis) {
                return jedis.get(key);
            }
        });
    }

    /**
     * @Description get T
     */
    public <T> T get(final String key, final T t) {
        return execute(new JedisCallback<T>() {
            @Override
            @SuppressWarnings("unchecked")
            public T doInRedis(Jedis jedis) {
                return (T) SerializeUtil.unserialize(jedis.get(getKeyBytes(key)));
            }
        });
    }

    /**
     * @Description 将key序列化
     */
    private byte[] getKeyBytes(String key) {
        return key.getBytes(RedisConst.DEFAULT_CHARSET);
    }

    /**
     * @Description 设置超时时间
     */
    public Long expire(final String key, final int seconds) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.expire(key, seconds);
            }
        });
    }

    /**
     * @Description 如有key已经存在，返回0，否则返回1
     */
    public Long hsetnx(final String key, final String field, final String value) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.hsetnx(key, field, value);
            }
        });
    }

    public Long hset(final String key, final String field, final String value) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.hset(key, field, value);
            }
        });
    }

    public Long hset(final String key, final String field, final Object value) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.hset(key.getBytes(), field.getBytes(), SerializeUtil.serialize(value));
            }
        });
    }

    /**
     * @Description 删除指定key的hash的多个fields;如果key存在且被删除，返回1，否则返回0
     */
    public Long hdel(final String key, final String... fields) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.hdel(key, fields);
            }
        });
    }

    public void hdel(final String key, final Iterable<String> fields) {
        execute(new JedisCallback<Object>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                for (String field : fields) {
                    pipeline.hdel(key, field);
                }
                pipeline.sync();
                return null;
            }
        });
    }

    public Map<String, String> hgetAll(final String key) {
        return execute(new JedisCallback<Map<String, String>>() {
            @Override
            public Map<String, String> doInRedis(Jedis jedis) {
                return jedis.hgetAll(key);
            }
        });
    }

    /**
     * @param key
     * @param clzss
     * @return
     */
    public <T> Map<String, T> hgetAll(final String key, Class<T> clzss) {
        return execute(new JedisCallback<Map<String, T>>() {
            @Override
            @SuppressWarnings("unchecked")
            public Map<String, T> doInRedis(Jedis jedis) {
                Map<byte[], byte[]> byteMap = jedis.hgetAll(key.getBytes());
                if (byteMap != null) {
                    Map<String, T> retMap = new HashMap<String, T>(byteMap.size());
                    for (Entry<byte[], byte[]> e : byteMap.entrySet()) {
                        retMap.put(new String(e.getKey()), (T) SerializeUtil.unserialize(e.getValue()));
                    }
                    return retMap;
                }
                return null;
            }
        });
    }

    /**
     * @param key
     * @param min
     * @param max
     * @return
     */
    public Set<String> zrangeByScore(final String key, final double min, final double max) {
        return execute(new JedisCallback<Set<String>>() {
            @Override
            public Set<String> doInRedis(Jedis jedis) {
                return jedis.zrangeByScore(key, min, max);
            }
        });
    }

    /**
     * @param key
     * @param min
     * @param max
     * @return
     */
    public Long zremrangeByScore(final String key, final double min, final double max) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.zremrangeByScore(key, min, max);
            }
        });
    }

    /**
     * @param key
     * @param score
     * @param member
     * @return
     */
    public Long zadd(final String key, final double score, final String member) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.zadd(key, score, member);
            }
        });
    }

    /**
     * @param key
     * @param member
     * @return
     */
    public Double zscore(final String key, final String member) {
        return execute(new JedisCallback<Double>() {
            @Override
            public Double doInRedis(Jedis jedis) {
                return jedis.zscore(key, member);
            }
        });
    }

    /**
     * @param key
     * @param members
     * @return
     */
    public long zrem(final String key, final String... members) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.zrem(key, members);
            }
        });
    }

    /**
     * @param key
     * @return
     */
    public Set<String> hkeys(final String key) {
        return execute(new JedisCallback<Set<String>>() {
            @Override
            public Set<String> doInRedis(Jedis jedis) {
                return jedis.hkeys(key);
            }
        });
    }

    /**
     * @param key
     * @param field
     * @return
     */
    public boolean hexists(final String key, final String field) {
        return execute(new JedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(Jedis jedis) {
                return jedis.hexists(key, field);
            }
        });
    }

    public String hget(final String key, final String field) {
        return execute(new JedisCallback<String>() {
            @Override
            public String doInRedis(Jedis jedis) {
                return jedis.hget(key, field);
            }
        });
    }

    /**
     * @param key
     * @param field
     * @param clazz
     * @return
     */
    public <T> T hget(final String key, final String field, Class<T> clazz) {
        return execute(new JedisCallback<T>() {
            @Override
            @SuppressWarnings("unchecked")
            public T doInRedis(Jedis jedis) {
                byte[] bytes = jedis.hget(key.getBytes(), field.getBytes());
                return null == bytes ? null : (T) SerializeUtil.unserialize(bytes);
            }
        });
    }

    /**
     * @Title: hmset
     * @param key
     * @param hashValue
     */
    public Object hmset(final String key, final Map<String, String> hashValue) {
        return execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                return jedis.hmset(key, hashValue);
            }
        });
    }

    public Object hmset(final String key, final Map<String, String> hashValue, final int expire) {
        return execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                jedis.hmset(key, hashValue);
                return jedis.expire(key, expire);

            }
        });
    }

    /**
     * @Description Return redis keys that matches the pattern string
     * @Title: keys
     * @param key
     * @param hashValue
     */
    public Set<String> keys(final String pattern) {
        return execute(new JedisCallback<Set<String>>() {
            @Override
            public Set<String> doInRedis(Jedis jedis) {
                return jedis.keys(pattern);
            }
        });
    }

    /**
     * @Description Add one or many members to set
     * @param key
     * @param members
     * @return
     */
    public Long sadd(final String key, final String... members) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.sadd(key, members);
            }
        });
    }

    /**
     * @Description Get all set members
     * @param key
     * @param members
     * @return
     */
    public Set<String> smembers(final String key) {
        return execute(new JedisCallback<Set<String>>() {
            @Override
            public Set<String> doInRedis(Jedis jedis) {
                return jedis.smembers(key);
            }
        });
    }

    /**
     * @Description Remove one or more members from a set
     * @param key
     * @param members
     * @return
     */
    public Long srem(final String key, final String... members) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.srem(key, members);
            }
        });
    }

    /**
     * @Description Get pipeline object
     * @return
     */
    public Pipeline pipelined() {
        return execute(new JedisCallback<Pipeline>() {
            @Override
            public Pipeline doInRedis(Jedis jedis) {
                return jedis.pipelined();
            }
        });
    }

    /**
     * @Description incr counter
     * @return
     */
    public long incrBy(final String key, final long num) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.incrBy(key, num);
            }
        });
    }

    /**
     * @Description decrBy
     * @return
     */
    public long decrBy(final String key, final long num) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.decrBy(key, num);
            }
        });
    }

    /**
     * @Description get counter
     * @return
     */
    public int getCounter(final String key) {
        return execute(new JedisCallback<Integer>() {
            @Override
            public Integer doInRedis(Jedis jedis) {
                try {
                    return Integer.valueOf(jedis.get(key));
                } catch (NumberFormatException e) {
                    return -1;
                }
            }
        });
    }

    /************* update index keys ********************/

    /**
     * @Description set and update index
     */
    public void indexSet(final String indexKey, final String key, final String value) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                pipeline.set(key, value);
                pipeline.sadd(indexKey, key);
                pipeline.sync();
                return null;
            }
        });
    }

    /**
     * @Description del and update index
     */
    public void indexDel(final String indexKey, final String key) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                pipeline.del(key);
                pipeline.srem(indexKey, key);
                pipeline.sync();
                return null;
            }
        });
    }

    /**
     * @Description hset and update index
     */
    public Long indexHset(final String indexKey, final String key, final String field, final String value) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> response = pipeline.hset(key, field, value);
                pipeline.sadd(indexKey, key);
                pipeline.sync();
                return response.get();
            }
        });
    }

    /**
     * @Description hset and update index
     */
    public Long indexHset(final String indexKey, final String key, final String field, final Object value) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> response = pipeline.hset(key.getBytes(), field.getBytes(),
                        SerializeUtil.serialize(value));
                pipeline.sadd(indexKey.getBytes(), key.getBytes());
                pipeline.sync();
                return response.get();
            }
        });
    }

    /**
     * @Description hdel and update index
     */
    public void indexHdel(final String indexKey, final String key, final String... fields) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                jedis.hdel(key, fields);
                int len = jedis.hlen(key).intValue();
                if (len == 0) {
                    jedis.srem(indexKey, key);
                }
                return null;
            }
        });
    }

    /**
     * @Description sadd and update index
     */
    public Long indexSadd(final String indexKey, final String key, final String... members) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> response = pipeline.sadd(key, members);
                pipeline.sadd(indexKey, key);
                pipeline.sync();
                return response.get();
            }
        });
    }

    /**
     * @Description zadd and update index
     */
    public Long indexZadd(final String indexKey, final String key, final double score, final String member) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> response = pipeline.zadd(key, score, member);
                pipeline.sadd(indexKey, key);
                pipeline.sync();
                return response.get();
            }
        });
    }

    /**
     * @Description zadd and update sortindex
     * @param expire
     *            key expire in second
     */
    public Long sortedIndexZadd(final String indexKey, final long expire, final String member) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.zadd(indexKey, System.currentTimeMillis() + expire * 1000, member);
            }
        });
    }

    /**
     * @Description: zremByScore < currentTimeMillis and update sortindex
     */
    public Long sortedIndexZremrangeByScore(final String indexKey) {
        return execute(new JedisCallback<Long>() {
            @Override
            public Long doInRedis(Jedis jedis) {
                return jedis.zremrangeByScore(indexKey, 0, System.currentTimeMillis());
            }
        });
    }

    /**
     * @Description sscan set members by key. default count 10
     */
    public Set<String> sscan(final String key) {
        return sscan(key, null, DEFAULT_COUNT);
    }

    /**
     * @Description sscan set members by key and count
     */
    public Set<String> sscan(final String key, final int count) {
        return sscan(key, null, count);
    }

    /**
     * @Description sscan set members by key and pattern. default count 10
     */
    public Set<String> sscan(final String key, final String pattern) {
        return sscan(key, pattern, DEFAULT_COUNT);
    }

    /**
     * @Description sscan set members by key and pattern and count
     */
    public Set<String> sscan(final String key, final String pattern, final int count) {
        return execute(new JedisCallback<Set<String>>() {
            @Override
            public Set<String> doInRedis(Jedis jedis) {
                boolean sscannDone = false;
                Set<String> resSet = new HashSet<String>(0);
                String cursor = "0";
                ScanParams params = new ScanParams();
                params.count(count);
                if (pattern != null) {
                    params.match(pattern);
                }
                while (!sscannDone) {
                    ScanResult<String> scanResults = jedis.sscan(key, cursor, params);
                    for (String key : scanResults.getResult()) {
                        resSet.add(key);
                    }
                    cursor = scanResults.getStringCursor();
                    if (cursor.equalsIgnoreCase("0")) {
                        sscannDone = true;
                    }
                }
                return resSet;
            }
        });
    }

    /**
     * @Description scan key by pattern. default count 10
     */
    public Set<String> scan(final String pattern) {
        return scan(pattern, DEFAULT_COUNT);
    }

    /**
     * @Description scan key by pattern and count
     */
    public Set<String> scan(final String pattern, final int count) {
        return execute(new JedisCallback<Set<String>>() {
            @Override
            public Set<String> doInRedis(Jedis jedis) {
                boolean scanDone = false;
                Set<String> resSet = new HashSet<String>(0);
                String cursor = "0";
                ScanParams params = new ScanParams();
                params.match(pattern);
                params.count(count);
                while (!scanDone) {
                    ScanResult<String> scanResults = jedis.scan(cursor, params);
                    for (String key : scanResults.getResult()) {
                        resSet.add(key);
                    }
                    cursor = scanResults.getStringCursor();
                    if (cursor.equalsIgnoreCase("0")) {
                        scanDone = true;
                    }
                }
                return resSet;
            }
        });
    }

    /**
     * @Description zscan SortedSet members by key. default count 10
     */
    public Set<String> zscan(final String key) {
        return zscan(key, null, DEFAULT_COUNT);
    }

    /**
     * @Description zscan SortedSet members by key and count.
     */
    public Set<String> zscan(final String key, final int count) {
        return zscan(key, null, count);
    }

    /**
     * @Description zscan SortedSet members by key and pattern. default count 10
     */
    public Set<String> zscan(final String key, final String pattern) {
        return zscan(key, pattern, DEFAULT_COUNT);
    }

    /**
     * @Description zscan SortedSet members by key and pattern and count
     */
    public Set<String> zscan(final String key, final String pattern, final int count) {
        return execute(new JedisCallback<Set<String>>() {
            @Override
            public Set<String> doInRedis(Jedis jedis) {
                boolean sscannDone = false;
                Set<String> resSet = new HashSet<String>(0);
                String cursor = "0";
                ScanParams params = new ScanParams();
                params.count(count);
                if (pattern != null) {
                    params.match(pattern);
                }
                while (!sscannDone) {
                    ScanResult<Tuple> scanResults = jedis.zscan(key, cursor, params);
                    for (Tuple tuple : scanResults.getResult()) {
                        resSet.add(tuple.getElement());
                    }
                    cursor = scanResults.getStringCursor();
                    if (cursor.equalsIgnoreCase("0")) {
                        sscannDone = true;
                    }
                }
                return resSet;
            }
        });
    }

    /**
     * @Description set and update index
     */
    public void indexSetBatchDelete(final String indexKey, final List<String> keys) {
        execute(new JedisCallback<Object>() {
            @Override
            public Object doInRedis(Jedis jedis) {
                if (null == keys || keys.size() == 0) {
                    return null;
                }
                Pipeline pipeline = jedis.pipelined();
                for (String key : keys) {
                    pipeline.del(key);
                }
                String[] indexValue = new String[keys.size()];
                indexValue = keys.toArray(indexValue);
                pipeline.srem(indexKey, indexValue);
                pipeline.sync();
                return null;
            }
        });
    }
}
